package com.atguigu.realtime.app.dwd.db;


import com.atguigu.realtime.app.BaseSqlApp;
import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;


/**
 * com.atguigu.realtime.app.dwd.db.Dwd_15_DwdInteractionComment
 * 互动域评价事务事实表（练习）
 *
 *  建立 MySQL-Lookup 字典表，读取评论表数据，关联字典表以获取评价（好评、中评、差评、自动），
 *  将结果写入 Kafka 评价主题。
 *      前文提到，与字典表关联时 ttl 的设置主要是考虑到从外部介质查询维度数据的时间，此处设置为 5s。
 *
 *      用户提交评论时评价表会插入一条数据，筛选操作类型为 insert 的数据即可。
 */
public class Dwd_15_DwdInteractionComment extends BaseSqlApp {
    public static void main(String[] args) {
        new Dwd_15_DwdInteractionComment().init(
                3015,2,
                "Dwd_15_DwdInteractionComment"
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env,
                          StreamTableEnvironment tEnv) {
        tEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(10));

        // 1. 读取ods_db
        readOdsDb(tEnv, "Dwd_11_DwdToolCouponGet");


        // TODO 2. 读取评论表数据
        Table commentInfo  = tEnv.sqlQuery("select\n" +
                "data['id'] id,\n" +
                "data['user_id'] user_id,\n" +
                "data['sku_id'] sku_id,\n" +
                "data['order_id'] order_id,\n" +
                "data['create_time'] create_time,\n" +
                "data['appraise'] appraise,\n" +
                "pt,\n" +
                "ts\n" +
                "from ods_db\n" +
                "where `database`='gmall2022' \n" +
                "and `table` = ''comment_info'\n" +
                "and `type` = 'insert'\n");
        tEnv.createTemporaryView("comment_info",commentInfo);

        // 3. 读取字典表
        readBaseDic(tEnv);

        Table result = tEnv.sqlQuery("select\n" +
                "ci.id,\n" +
                "ci.user_id,\n" +
                "ci.sku_id,\n" +
                "ci.order_id,\n" +
                "date_format(ci.create_time,'yyyy-MM-dd') date_id,\n" +
                "ci.create_time,\n" +
                "ci.appraise,\n" +
                "dic.dic_name,\n" +
                "ts\n" +
                "from comment_info ci \n" +
                "join base_dic for system_time as of ci.pt as dic on ci.appraise=dic.code");


        tEnv.executeSql("create table dwd_interaction_comment(\n" +
                "id string,\n" +
                "user_id string,\n" +
                "sku_id string,\n" +
                "order_id string,\n" +
                "date_id string,\n" +
                "create_time string,\n" +
                "appraise_code string,\n" +
                "appraise_name string,\n" +
                "ts bigint \n" +
                ")" + SQLUtil.getKafkaSink(Constant.TOPIC_DWD_INTERACTION_COMMENT));

        result.executeInsert("dwd_interaction_comment");
    }
}

/*

 */